package net.oschina.allchat.server.framework.pipe.parsing;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;

import net.oschina.allchat.message.processing.convention.NamingConventionParserFactory;
import net.oschina.allchat.message.processing.json.ParsingFactory;
import net.oschina.allchat.message.processing.parsers.SimpleObjectParserFactory;
import net.oschina.allchat.message.processing.parsers.core.stanza.IqParserFactory;
import net.oschina.allchat.message.processing.parsers.error.ErrorParserFactory;
import net.oschina.allchat.message.processing.parsers.error.StanzaErrorDetailsParserFactory;
import net.oschina.allchat.message.processing.parsing.IParserFactory;
import net.oschina.allchat.message.processing.parsing.IParsingFactory;
import net.oschina.allchat.protocol.core.IError;
import net.oschina.allchat.protocol.core.Protocol;
import net.oschina.allchat.protocol.core.ProtocolChain;
import net.oschina.allchat.protocol.core.ProtocolException;
import net.oschina.allchat.protocol.core.stanza.Iq;
import net.oschina.allchat.protocol.core.stanza.Stanza;
import net.oschina.allchat.protocol.core.stanza.error.InternalServerError;
import net.oschina.allchat.protocol.core.stanza.error.StanzaError;
import net.oschina.allchat.protocol.core.stream.Stream;
import net.oschina.allchat.protocol.core.stream.error.InvalidFrom;
import net.oschina.allchat.protocol.core.stream.error.InvalidXml;
import net.oschina.allchat.server.framework.core.annotations.Component;
import net.oschina.allchat.server.framework.core.commons.osgi.IBundleContextAware;
import net.oschina.allchat.server.framework.core.commons.osgi.IContributionTracker;
import net.oschina.allchat.server.framework.core.commons.osgi.OsgiUtils;
import net.oschina.allchat.server.framework.core.commons.utils.CommonUtils;
import net.oschina.allchat.server.framework.core.config.IConfiguration;
import net.oschina.allchat.server.framework.core.config.IConfigurationAware;
import net.oschina.allchat.server.framework.core.connection.IConnectionContext;
import net.oschina.allchat.server.framework.core.integration.IMessage;
import net.oschina.allchat.server.framework.core.integration.IMessageProcessor;
import net.oschina.allchat.server.framework.core.repository.IInitializable;

import org.osgi.framework.Bundle;
import org.osgi.framework.BundleContext;

@Component("minimum.message.parsing.processor")
public class MinimumMessageParsingProcessor implements IMessageProcessor, IBundleContextAware,
			IContributionTracker, IInitializable, IConfigurationAware {
	private static final String SEPARATOR_OF_PROTOCOLS = "->";
	private static final String SEPARATOR_OF_LOCALNAME_NAMESPACE = "|";
	private static final String VALUE_NULL = "null";
	private static final String PROPERTY_NAME_PROTOCOL_CHAIN = "protocol-chain";
	private static final String PROPERTY_NAME_TYPE = "type";
	private static final String PROPERTY_NAME_CLASS = "class";
	private static final String PROPERTY_NAME_PARSER_FACTORY = "parser-factory";
	private static final String TYPE_SIMPLE = "simple";
	private static final String TYPE_NAMING_CONVENTION = "naming-convention";
	private static final String TYPE_CUSTOM = "custom";
	private static final String SEPARATOR_PARSERS = ",";
	private static final String KEY_ALLCHAT_PARSERS = "AllChat-Parsers";
	
	protected IParsingFactory parsingFactory;
	protected BundleContext bundleContext;
	protected Map<Bundle, List<ProtocolChain>> bundleAndProtocolChains;
	
	private boolean stanzaErrorAttachOriginalMessage;
	
	public MinimumMessageParsingProcessor() {
		parsingFactory = new ParsingFactory();
		bundleAndProtocolChains = new HashMap<Bundle, List<ProtocolChain>>();
	}
	
	@Override
	public void init() {
		registerPredefinedParsers();
		trackContributedParsers();
	}
	
	protected void registerPredefinedParsers() {
		parsingFactory.register(
				ProtocolChain.first(Iq.PROTOCOL),
				new IqParserFactory()
		);
		parsingFactory.register(
				ProtocolChain.first(StanzaError.PROTOCOL),
				new ErrorParserFactory<StanzaError>(IError.Type.STANZA)
		);
		parsingFactory.register(
				ProtocolChain.first(StanzaError.PROTOCOL).next(
						StanzaError.PROTOCOL_ERROR_DEFINED_CONDITION),
				new StanzaErrorDetailsParserFactory()
		);
	}

	protected void trackContributedParsers() {
		OsgiUtils.trackContribution(bundleContext, KEY_ALLCHAT_PARSERS, this);
	}

	@Override
	public void process(IConnectionContext context, IMessage message) {
		Object out;
		try {
			Object object = parsingFactory.parse((String)message.getPayload());
			if (object instanceof Stanza) {
				Stanza stanza = (Stanza)object;
				
				if (stanza.getFrom() == null) {
					stanza.setFrom(context.getId());
				}
				
				// (rfc3920 9.1.2)
				if (!stanza.getFrom().equals(context.getId())) {
					throw new ProtocolException(new InvalidFrom());
				}
			} else if (object instanceof Stream) {
				Stream stream = (Stream)object;
				if (!Boolean.TRUE.equals(stream.getClose())) {
					throw new ProtocolException(new InvalidXml("open stream not accepted"));
				}
			}
			
			out = object;
			
		} catch (ProtocolException e) {
			IError error = e.getError();
			if ((error instanceof StanzaError) && stanzaErrorAttachOriginalMessage) {
				((StanzaError)error).setOriginalMessage((String)message.getPayload());
			}
			
			out = error;
		} catch (RuntimeException e) {
			out = new InternalServerError(e.getMessage());
		}
		
		context.write(out);
	}

	@Override
	public void setBundleContext(BundleContext bundleContext) {
		this.bundleContext = bundleContext;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public void found(Bundle bundle, String contribution) throws Exception {
		StringTokenizer tokenizer = new StringTokenizer(contribution, SEPARATOR_PARSERS);
		
		List<ProtocolChain> protocolChains = new ArrayList<ProtocolChain>();
		while (tokenizer.hasMoreTokens()) {
			String parserString = tokenizer.nextToken();
			
			Map<String, String> properties = CommonUtils.parsePropertiesString(
					parserString,
					new String[] {
						PROPERTY_NAME_CLASS,
						PROPERTY_NAME_TYPE,
						PROPERTY_NAME_PROTOCOL_CHAIN,
						PROPERTY_NAME_PARSER_FACTORY
					}
			);
			
			String sProtocolChain = properties.get(PROPERTY_NAME_PROTOCOL_CHAIN);
			ProtocolChain protocolChain = parseProtocolChain(sProtocolChain);
			
			String sType = properties.get(PROPERTY_NAME_TYPE);
			if (sType == null) {
				sType = TYPE_CUSTOM;
			}
			
			IParserFactory<?> parserFactory;
			if (TYPE_CUSTOM.equals(sType)) {
				String sParserFactory = properties.get(PROPERTY_NAME_PARSER_FACTORY);
				if (sParserFactory == null)
					throw new IllegalArgumentException("null parser factory[register parser]");
				
				Class<?> parserFactoryClass = bundle.loadClass(sParserFactory);
				
				if (!(IParserFactory.class.isAssignableFrom(parserFactoryClass)))
					throw new RuntimeException(String.format("%s must implement %s[register parser]",
							parserFactoryClass, IParserFactory.class));
					
				parserFactory = (IParserFactory<?>)parserFactoryClass.newInstance();
			} else {
				String sClass = properties.get(PROPERTY_NAME_CLASS);
				if (sClass == null)
					throw new IllegalArgumentException("null class[register parser]");
				
				Class<?> clazz = bundle.loadClass(sClass);
				
				Protocol protocol = protocolChain.get(protocolChain.size() - 1);
				if (TYPE_SIMPLE.equals(sType)) {
					parserFactory = new SimpleObjectParserFactory(protocol, clazz);
				} else if (TYPE_NAMING_CONVENTION.equals(sType)) {
					parserFactory = new NamingConventionParserFactory(protocol, clazz);
				} else {
					throw new RuntimeException(String.format("unknown parser type %s", sType));
				}
			}
			
			parsingFactory.register(protocolChain, parserFactory);
			protocolChains.add(protocolChain);
		}
		
		bundleAndProtocolChains.put(bundle, protocolChains);
	}

	private ProtocolChain parseProtocolChain(String sProtocolChain) {
		StringTokenizer tokenizer = new StringTokenizer(sProtocolChain, SEPARATOR_OF_PROTOCOLS);
		
		ProtocolChain protocolChain = null;
		while (tokenizer.hasMoreTokens()) {
			String sProtocol = tokenizer.nextToken();
			
			String localName;
			String namespace;
			int seperator = sProtocol.indexOf(SEPARATOR_OF_LOCALNAME_NAMESPACE);
			if (seperator == -1) {
				localName = sProtocol.trim();
				namespace = VALUE_NULL;
			} else {
				localName = sProtocol.substring(0, seperator).trim();
				namespace = sProtocol.substring(seperator + 1, sProtocol.length()).trim();
			}
			
			if (localName.length() == 0 || namespace.length() == 0) {
				throw new IllegalArgumentException(String.format("invalid protocol chain[%s]", sProtocolChain));
			}
			
			if (VALUE_NULL.equals(namespace)) {
				namespace = null;
			}
			
			Protocol protocol = new Protocol(namespace, localName);
			if (protocolChain == null) {
				protocolChain = ProtocolChain.first(protocol);
			} else {
				protocolChain = protocolChain.next(protocol);
			}
		}
		
		return protocolChain;
	}

	@Override
	public void lost(Bundle bundle, String contribution) throws Exception {
		List<ProtocolChain> protocolChains = bundleAndProtocolChains.remove(bundle);
		
		if (protocolChains != null) {
			for (ProtocolChain protocolChain : protocolChains) {
				parsingFactory.unregister(protocolChain);
			}
		}
	}

	@Override
	public void setConfiguration(IConfiguration configuration) {
		stanzaErrorAttachOriginalMessage = configuration.getBoolean("stanza.error.attach.original.message", false);
	}

}
